package spring.cloud.tasks.executor.executor.task;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.utils.CloseableExecutorService;
import spring.cloud.tasks.common.ExecutorServiceUtil;
import spring.cloud.tasks.common.ThreadFactory;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Setter
@Getter
public class TaskExecutorNodeCacheOrTreeCacheManager {
    private Map<String/* path */, NodeCache> pathToNodeCacheMap = new HashMap<>();
    private Map<String/* path-depth */, TreeCache> pathAndDepthToTreeCacheMap = new HashMap<>();
    private String executorId;
    private String taskId;
    private CuratorFramework curatorFramework;
    private ExecutorService executorService;

    public TaskExecutorNodeCacheOrTreeCacheManager(CuratorFramework curatorFramework, String executorId, String taskId) {
        this.executorId = executorId;
        this.taskId = taskId;
        this.curatorFramework = curatorFramework;
        this.executorService = Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-" + taskId + "-watcher", false));
    }

    private static String buildPathAndDepthKey(String path, int depth) {
        return path + "-" + depth;
    }

    public void addNodeCacheListener(final String path, final NodeCacheListener nodeCacheListener) {
        synchronized (this) {
            NodeCache nodeCache = buildNodeCache(path);
            if (nodeCache != null) {
                nodeCache.getListenable().addListener(nodeCacheListener);
            }
        }
    }

    public NodeCache buildNodeCache(String path) {
        synchronized (this) {
            try {
                NodeCache nodeCache = pathToNodeCacheMap.get(path);
                if (nodeCache != null) {
                    return nodeCache;
                }
                nodeCache = new NodeCache(curatorFramework, path);
                nodeCache.start();//开启成功后再put
                pathToNodeCacheMap.put(path, nodeCache);
                return nodeCache;
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                return null;
            }
        }
    }

    public void closeNodeCache(String path) {
        synchronized (this) {
            NodeCache nodeCache = pathToNodeCacheMap.remove(path);
            if (nodeCache != null) {
                try {
                    nodeCache.close();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    }

    //
    public void addTreeCacheListener(final String path, final int depth, final TreeCacheListener treeCacheListener) {
        synchronized (this) {
            TreeCache treeCache = buildTreeCache(path, depth);
            if (treeCache != null) {
                treeCache.getListenable().addListener(treeCacheListener);
            }
        }
    }

    private TreeCache buildTreeCache(String path, int depth) {
        synchronized (this) {
            try {
                String pathAndDepth = buildPathAndDepthKey(path, depth);
                TreeCache treeCache = pathAndDepthToTreeCacheMap.get(pathAndDepth);
                if (treeCache != null) {
                    return treeCache;
                }
                treeCache = TreeCache.newBuilder(curatorFramework, path).setMaxDepth(depth).setExecutor(new CloseableExecutorService(executorService, false)).build();
                treeCache.start();//开启成功后再put
                pathAndDepthToTreeCacheMap.put(pathAndDepth, treeCache);
                return treeCache;
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                return null;
            }
        }
    }

    public void closeTreeCache(String path, int depth) {
        synchronized (this) {
            String pathAndDepth = buildPathAndDepthKey(path, depth);
            TreeCache treeCache = pathAndDepthToTreeCacheMap.remove(pathAndDepth);
            if (treeCache != null) {
                try {
                    treeCache.close();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    }

    /**
     * add by huihui
     * {@link spring.cloud.tasks.executor.executor.task.TaskExecutor#shutdown()}
     * 注意如果是任务删除和任务停止这种事件。
     * 处理必须是新启动一个线程进行处理。
     * 因为任务删除和任务停止这种事件所处的线程就是当前绑定的线程池。
     * 不能自己把自己清理掉，会造成当前任务的资源没有清理干净而线程提前关闭。
     */
    public void shutdown() {
        synchronized (this) {
            closeNodeCaches();
            closeTreeCaches();
            //
            ExecutorServiceUtil.shutdownNow(executorService);
        }
    }

    public void closeNodeCaches() {
        synchronized (this) {
            Iterator<Entry<String, NodeCache>> iterator = pathToNodeCacheMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, NodeCache> next = iterator.next();
                NodeCache nodeCache = next.getValue();
                try {
                    iterator.remove();
                    nodeCache.close();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    }

    public void closeTreeCaches() {
        synchronized (this) {
            Iterator<Entry<String, TreeCache>> iterator = pathAndDepthToTreeCacheMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, TreeCache> next = iterator.next();
                TreeCache treeCache = next.getValue();
                try {
                    iterator.remove();
                    treeCache.close();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    }
}
